package com.xzx.flink.streamapi.processfunction;

import com.xzx.flink.bean.ClickEvent;
import com.xzx.flink.streamapi.source.ClickSource;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;

/**
 * @version 1.0
 * @auther xinzhixuan
 * @date 2022/5/4 13:02
 */
public class ProcessFunction_01_Test {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        env.addSource(new ClickSource())
                .assignTimestampsAndWatermarks(WatermarkStrategy.<ClickEvent>forMonotonousTimestamps()
                        .withTimestampAssigner(new SerializableTimestampAssigner<ClickEvent>() {
                            @Override
                            public long extractTimestamp(ClickEvent element, long recordTimestamp) {
                                return element.getTimestamp();
                            }
                        }))
                .process(new ProcessFunction<ClickEvent, String>() {
                    @Override
                    public void processElement(ClickEvent value, ProcessFunction<ClickEvent, String>.Context ctx, Collector<String> out) throws Exception {
                        //Mary输出一次，Bob输出2次，其他直接过滤掉，把map flatmap filter
                        if (value.getUser().equals("Mary")) {
                            out.collect(value.getUser());
                        } else if (value.getUser().equals("Bob")) {
                            out.collect(value.getUser());
                            out.collect(value.getUser());
                        }
                        // 因为是事件时间语义，所以可以获取水位线
                        System.out.println(ctx.timerService().currentWatermark());
                    }
                })
                .print();

        env.execute(ProcessFunction_01_Test.class.getSimpleName());
    }
}
